<?php

namespace app\cron\controller;
use  app\common\controller\DBase;
use app\systems\logic\SystemMqLogic;
use app\systems\logic\MqGroupLogic;
use app\systems\logic\MqTaskServiceLogic;
use app\data\util\Rabbitmq;
use think\App;


class Crontab
{
    /**
     * 日志文件地址
     * @var unknown
     */
    protected $_logFile = null;

    protected $rediss;

    protected $rabbitmq;



    /**
     * 常驻进程：执行分组id的队列任务,一个分组一个常驻进程
     * @author xx
     * @Date 2024/3/12 17:13
     * @param $group_id
     * php app/index.php cron/crontab/real_process 1
     * cron/crontab/real_process 2
     * cron/crontab/real_process 3
     */
    public function real_process($group_id=''){
        $mqConfig = MqGroupLogic::get_mq_group_by_ids($group_id);
        $rabbitQq = new Rabbitmq($mqConfig['mq_config']);
        $rabbitQq->task_start_time = time();


        $rabbitQq->mq_consume_queue($mqConfig['queue_name'], function($envelope, $queue) use ($group_id,$mqConfig,$rabbitQq) {
            if (time() - $rabbitQq->task_start_time > $rabbitQq->task_execute_time) exit("任务已执行20分钟,等待下次执行");
            try
            {
                $queue_name = $mqConfig['queue_name'];
                $message = $envelope->getBody ();
                $message = json_decode($message,true);
                if(!is_null($message)){
                    $failureCount = isset($message['failure_count']) ? (int)$message['failure_count'] : 0;
                    $obj = new MqTaskServiceLogic();
                    $queue_funtion = !empty($mqConfig['queue_function'])?$mqConfig['queue_function']:$mqConfig['queue_name'];
                    $result = call_user_func_array([$obj,$queue_funtion],[$message]);
                    $code = $result['code'] ?? 0;
                    if ($code == 20048) {
                        $queue->nack($envelope->getDeliveryTag());
                    } else {
                        $status = $result['status'] ?? 0;
                        ## 这里成功逻辑调整为：true 或 $result['status'] = 1 20210715-cyz
                        if ($result === true || $status == 1) {
                            $queue->ack( $envelope->getDeliveryTag());
                        }else{
                            #### 执行失败的，在重试次数内重新推送回
                            if($failureCount < SystemMqLogic::TASK_MAX_RETRY_TIMES)
                            {
                                $failureCount++;
                                $message['failure_count'] = $failureCount;
                                $flag = MqGroupLogic::mq_send_message_by_id($rabbitQq,$group_id,$message);
                                sleep(1);
                               // file_put_contents($this->_logFile, sprintf('ERROR -- %s  将失败的任务重新推送会消息队列，推送结果：%s' . PHP_EOL, date('Y-m-d H:i:s'), var_export($flag, true)), FILE_APPEND);
                                if ($flag)
                                    $queue->ack($envelope->getDeliveryTag());
                                else
                                    $queue->nack($envelope->getDeliveryTag());
                            }
                            else
                            {
                                //@TODO 处理失败的任务处理逻辑
                                $queue->ack($envelope->getDeliveryTag());
                            }
                        }
                    }
                }
                else
                {
                    //无效的消息，直接应答成功
                    $queue->ack($envelope->getDeliveryTag());
                }
            }
            catch (\Exception $e)
            {
                //代码异常
                $queue->ack($envelope->getDeliveryTag());
            }
        }, false, true);
    }


    /**
     * 队列任务组检测调度,每分钟检测一次
     * @author xx
     * @Date 2024/3/12 17:13
     * @throws Exception
     * php appdal/index.php cron/crontab/real_time
     */
    public function real_time(){
        ####1、获取队列列表
        $cron_groups =  (new SystemMqLogic())->get_mq_group_list();
        ####3、根据消费总数和进程执行情况处理调度
        ##1、当队列待消费数为0时，如果进程仍在执行，则关闭
        ##2、当队列待消费数不为0时，根据总消费数开启多进程消费数据
        if (is_array($cron_groups) && !empty($cron_groups)) {
            ////队列进程调度逻辑
            foreach ($cron_groups as $group) {
                try{
                    $rabbit_mq = new Rabbitmq($group['mq_config']);

                    $group_id = $group['id']; //任务id
                    $queue_name = $group['queue_name'];  //队列名称
                    ####根据待消费总数、最大进程数、单进程处理能力数、现有进程数 调度生成进程数
                    $max_process_num = $group['max_process_num']; ##最大进程数
                    $max_do_ready_num = $group['max_do_ready_num'];  ##单进程处理能力数
                    $mq_reday_count = $rabbit_mq->mq_get_queue_num($queue_name);  ##待消费总数
                    $shell = "ps -ef | grep -E '/www/server/php/80/bin/php think realProcess ".$group_id."\>' | grep -v grep | wc -l";
                    $process_count_now = (int)shell_exec($shell);  ##现有进程数
                    if ($mq_reday_count){   #待消费数不为0
                        ####根据待消费总数、最大进程数、单进程处理能力数、现有进程数 调度生成进程数
                        $process_need_count = ceil($mq_reday_count/$max_do_ready_num);  ##需要的进程数

                        if ( $process_count_now >= $process_need_count || $process_count_now >= $max_process_num) continue;      ##如果现有进程满足需要的进程数，则退出判断

                        ####否则开启多进程
                        $process_need_count = $process_need_count - $process_count_now;
                        $process_need_count = min($process_need_count,$max_process_num);   ##最大开启进程数
                        $php_path = '/www/server/php/80/bin/php';
                        while ($process_need_count){
                            $child = new \Swoole\Process(function (\Swoole\Process $process) use ($group_id,$php_path) {
                                swoole_set_process_name(sprintf('php-ps-process:%s',$group_id));
                                $process->exec($php_path, ['think', 'realProcess', $group_id]);
                            } ,false);
                            $child_id = $child->start(); //启动子进程
                            $process_need_count--;
                        }
//                    while ($res = swoole_process::wait()) {
//                        log_message('ERROR','========================='.var_export($res,true));
//                        if (!$res) {continue;}
//                    }
                    }
                    else {
                        # 待消费数为空，如果还有进程在执行中，就关闭对应进程
                        echo $process_count_now;
                        if ($process_count_now > 0){
                            sleep(2);
                            //执行批量关闭进程
                            $shell = "ps -ef | grep -E '/www/server/php/80/bin/php think realProcess ".$group_id."\>' | grep -v grep | awk '{print $2}' | xargs kill -9";
                            shell_exec($shell);
                            echo  '队列id'.$group_id.'待消费数据为空，运行进程已被关闭';
                        }else{
                            echo  '暂时无数据,无进程。'."\n";
                        }
                    }
                }catch (\Exception $e){
                    echo $e->getMessage();
                    continue;
                }
            }
        }

    }

    /**
     * 检查对应进程是否在执行
     * @author xx
     * @Date 2024/3/12 17:13
     */
    public function test_process(){

        $shell = 'php /mnt/ybcloud_ebay/appdal/index.php /cron/crontab/real_process 1';
        $shell = 'php /mnt/ybcloud_ebay/appdal/index.php /cron/crontab/real_process 1 > /dev/null 2>&1 & echo $!';
//        $shell = 'ps -ef | grep "/cron/crontab/real_process 1" | grep -v grep | wc -l';
        $shell = 'ps -ef | grep "/cron/crontab/real_process 1" | grep -v grep';
        $shell = 'ps -ef | grep "/cron/crontab/real_process 1" | grep -v grep > /dev/null 2>&1 & echo $!';

        $shell = "ps -ef | grep -E '/cron/crontab/real_process 1' | grep -v grep | awk '{print $2}' | xargs kill -9";

        $shell = 'ps -ef | grep -E "/cron/crontab/real_process 1" | grep -v grep | cut -c 9-15 | xargs kill -9';

        $output = shell_exec($shell);
        var_dump($output);

    }










}